package cn._51doit.day05;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @create: 2021-10-20 11:18
 * @author: 今晚打脑斧
 * @program: CountWindowAllDemo
 * @Description:
 *
 * 不keyBy，直接划分回话窗口，即按照ProcessingTime划分回话窗口
 * 底层调用的是windowAll方法，传入的是ProcessingTime类型的SessionWindow
 *
 * 如果使用的是ProcessingTime，就是当前系统时间 - 进入到该窗口中最后一条数据对于的系统时间 > 指定的时间间隔 ，前面的数据就会形成一个窗口触发
 *
 **/
public class ProcessingTimeSessionWindowAllDemo {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Integer> nums = lines.map(Integer::parseInt);

        /**
         * 触发条件就是最后一条数据传入的时间 减去 系统当前时间 大于指定的时间间隔
         * 数据是增量式计算
         */
        //按照时间间隔划分窗口
        AllWindowedStream<Integer, TimeWindow> windowedStream = nums.windowAll(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));
        SingleOutputStreamOperator<Integer> res = windowedStream.sum(0);

        res.print();
        env.execute();

    }
}
